import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

public class hbaseStu {

    /*
     * 不强制性创建表
     *
     * @tableName 表名
     * @family 列族列表
     * @config　配置信息
     */
    public static void creatTable(String tableName, String[] family, Configuration config) throws Exception{
        try (Connection connection = ConnectionFactory.createConnection(config);
             Admin admin = connection.getAdmin()) {
            HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));
            for (int i = 0; i < family.length; i++) {
                desc.addFamily(new HColumnDescriptor(family[i]));
            }
            if (admin.tableExists(desc.getTableName())) {
                System.out.println("table Exists!");
                throw new Exception("table Exists!");
            } else {
                admin.createTable(desc);
                System.out.println("create table Success!");
            }
        }
    }

    /*
     *  删表
     *  @tableName 表名
     *  @config 配置信息
     *
     */
    public static void deleteTable(String tableName, Configuration config) throws Exception {
        try (Connection connection = ConnectionFactory.createConnection(config);
             Admin admin = connection.getAdmin()) {
            TableName tn = TableName.valueOf(tableName);
            if (admin.tableExists(tn)) {
                admin.disableTable(tn);
                admin.deleteTable(tn);
            }
        }
    }

    /*
     * 插入数据
     *
     * @tableName　表名
     * @config 配置信息
     * @rowkey 行key
     * @colFamily 列族
     * @col　子列
     * @val　值
     *
     * */
    public static void instertRow(String tableName, Configuration config, String rowkey, String colFamily, String col, String val) throws Exception {
        try (Connection connection = ConnectionFactory.createConnection(config)) {
            Table table = connection.getTable(TableName.valueOf(tableName));
            Put put = new Put(Bytes.toBytes(rowkey));
            put.addColumn(Bytes.toBytes(colFamily), Bytes.toBytes(col), Bytes.toBytes(val));
            table.put(put);

            //批量插入
               /* List<Put> putList = new ArrayList<Put>();
                puts.add(put);
                table.put(putList);*/
            table.close();
            System.out.printf("adding success!!Table:%s,Row:%s,Column=%s:%s,Value=%s\n", tableName, rowkey, colFamily, col, val);
        }
    }

    /*
     * 根据rowkey查找数据
     *
     * @tableName　表名
     * @config 配置信息
     * @rowkey 行key
     * @colFamily 列族
     * @col　子列
     *
     * */
    public static Result getData(String tableName, Configuration config, String rowkey, String colFamily, String col) throws Exception {
        try (Connection connection = ConnectionFactory.createConnection(config)) {
            Table table = connection.getTable(TableName.valueOf(tableName));
            Get get = new Get(Bytes.toBytes(rowkey));
            if (colFamily != null && col == null)
                get.addFamily(Bytes.toBytes(colFamily));
            if (colFamily != null && col != null)
                get.addColumn(Bytes.toBytes(colFamily), Bytes.toBytes(col));
            Result result = table.get(get);
            table.close();
            return result;
        }
    }

    public static Result getData(String tableName, Configuration config, String rowkey, String colFamily) throws Exception {
        return getData(tableName, config, rowkey, colFamily, null);
    }

    public static Result getData(String tableName, Configuration config, String rowkey) throws Exception {
        return getData(tableName, config, rowkey, null, null);
    }

    /*
     * 批量查找数据
     * @table 表名
     * @config配置文件
     * @startRow 开始的行key
     * @stopRow　停止的行key
     *
     * hbase会将自己的元素按照key的ASCII码排序
     * 找出5193开头的元素
     *
     *   5193:1
     *   5193:2
     *   5194:1
     *   51939:1
     *   51942:1
     *
     *  scan.setStartRow("5193:#");
     *  scan.setStopRow("5193::");
     *
     *  原因：ASCII排序中："#" < "0-9" < ":"
     *  取出来的将是5193:后面跟着数字的元素
     * */
    public static List<Result> scanData(String tableName, Configuration config, String startRow, String stopRow, int limit) throws Exception {
        try (Connection connection = ConnectionFactory.createConnection(config)) {
            Table table = connection.getTable(TableName.valueOf(tableName));
            Scan scan = new Scan();
            if (startRow != null && stopRow != null) {
                scan.setStartRow(Bytes.toBytes(startRow));
                scan.setStopRow(Bytes.toBytes(stopRow));
            }
            scan.setBatch(limit);
            List<Result> result = new ArrayList<Result>();
            ResultScanner resultScanner = table.getScanner(scan);
            for (Result r : resultScanner) {
                result.add(r);
            }
            table.close();
            return result;
        }
    }

    public static List<Result> scanData(String tableName, Configuration config, int limit) throws Exception {
        return scanData(tableName, config, null, null, limit);
    }

    /*
     * 打印表
     * @tables 打印的表描述对象
     *
     * */
    public static void printTables(HTableDescriptor[] tables) {
        for (HTableDescriptor t : tables) {
            HColumnDescriptor[] columns = t.getColumnFamilies();
            System.out.printf("tables:%s,columns-family:\n", t.getTableName());
            for (HColumnDescriptor column : columns) {
                System.out.printf("\t%s\n", column.getNameAsString());
            }
        }
    }


    /*
     * 格式化输出
     * @result 结果
     *
     * */

    public static void showCell(Result result) {
        Cell[] cells = result.rawCells();
        for (Cell cell : cells) {
            System.out.println("RowName:" + new String(CellUtil.cloneRow(cell)) + " ");
            System.out.println("Timetamp:" + cell.getTimestamp() + " ");
            System.out.println("column Family:" + new String(CellUtil.cloneFamily(cell)) + " ");
            System.out.println("row Name:" + new String(CellUtil.cloneQualifier(cell)) + " ");
            System.out.println("value:" + new String(CellUtil.cloneValue(cell)) + " ");
            System.out.println("---------------");
        }
    }

    public static void main(String... args) {
        Configuration config = HBaseConfiguration.create();
        String tablename = "students";
        String[] column_family = {"Description","Courses","Home"};
        try {
            //创建表
            System.out.println("\n（1）创建讲义中的students表");
            creatTable(tablename, column_family, config);

            //插入行
            instertRow(tablename, config, "001", column_family[0], "Name", "Li Lei");
            instertRow(tablename, config, "001", column_family[0], "Height", "176");
            instertRow(tablename, config, "001", column_family[1], "Chinese", "80");
            instertRow(tablename, config, "001", column_family[1], "Math", "80");
            instertRow(tablename, config, "001", column_family[1], "Physics", "95");
            instertRow(tablename, config, "001", column_family[2], "Province", "Zhejiang");

            instertRow(tablename, config, "002", column_family[0], "Name", "Han Meimei");
            instertRow(tablename, config, "002", column_family[0], "Height", "183");
            instertRow(tablename, config, "002", column_family[1], "Chinese", "88");
            instertRow(tablename, config, "002", column_family[1], "Math", "77");
            instertRow(tablename, config, "002", column_family[1], "Physics", "66");
            instertRow(tablename, config, "002", column_family[2], "Province", "Beijing");

            instertRow(tablename, config, "003", column_family[0], "Name", "Xiao Ming");
            instertRow(tablename, config, "003", column_family[0], "Height", "162");
            instertRow(tablename, config, "003", column_family[1], "Chinese", "90");
            instertRow(tablename, config, "003", column_family[1], "Math", "90");
            instertRow(tablename, config, "003", column_family[1], "Physics", "90");
            instertRow(tablename, config, "003", column_family[2], "Province", "Shanghai");


            //扫描表
            System.out.println("\n（2）扫描创建后的students表");
            List<Result> results = scanData(tablename, config, 3);
            for (Result r : results) {
                showCell(r);
            }

            //查询
            System.out.println("\n（3）查询学生来自的省");
            Result result = getData(tablename, config, "001", column_family[2],"Province");
            showCell(result);

            //增加新列
            System.out.println("\n（4）增加新的列Courses:English，并添加数据");
            instertRow(tablename, config, "001", column_family[1], "English", "90");
            instertRow(tablename, config, "002", column_family[1], "English", "90");
            instertRow(tablename, config, "003", column_family[1], "English", "90");

            //增加新列族和新列
            System.out.println("\n（5）增加新的列族Contact和新列Contact:Email，并添加数据");
            Connection conn=ConnectionFactory.createConnection(config);
            Admin admin = conn.getAdmin();
            HColumnDescriptor newF = new HColumnDescriptor("Contact");
            admin.addColumnFamily(TableName.valueOf(tablename),newF);
            instertRow(tablename, config, "001", "Contact", "Email", "001@126.com");
            instertRow(tablename, config, "002", "Contact", "Email", "002@126.com");
            instertRow(tablename, config, "003", "Contact", "Email", "003@126.com");

            // (6）删除students表
            System.out.println("\n（6）删除students表");
            deleteTable(tablename,config);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }







}
